package com.xiaoyu.im.client.proxy;

import android.util.Log;

import androidx.annotation.IntDef;
import androidx.annotation.NonNull;
import androidx.annotation.Nullable;

import com.netease.nimlib.sdk.msg.model.CustomNotification;
import com.netease.nimlib.sdk.msg.model.IMMessage;
import com.xiaoyu.base.DBCall;
import com.xiaoyu.base.app.GlobalUI;
import com.xiaoyu.base.data.UserData;
import com.xiaoyu.base.entity.MessageEntity;
import com.xiaoyu.base.utils.ListUtil;
import com.xiaoyu.base.utils.RxUtils;
import com.xiaoyu.im.BuildConfig;
import com.xiaoyu.im.IMCallback;
import com.xiaoyu.im.IMLog;
import com.xiaoyu.im.IMUtil;
import com.xiaoyu.im.client.IMClient;
import com.xiaoyu.im.data.IMTokenData;
import com.xiaoyu.im.datamodel.AppMessage;
import com.xiaoyu.im.datamodel.ConversationWithMessageListWrapper;
import com.xiaoyu.im.datamodel.IMVendorType;
import com.xiaoyu.im.datamodel.ReceiveType;
import com.xiaoyu.im.datamodel.auth.IMTokenInfo;
import com.xiaoyu.im.datamodel.bridge.MessageBridge;
import com.xiaoyu.im.datamodel.common.IMConstant;
import com.xiaoyu.im.datamodel.exception.IMException;
import com.xiaoyu.im.datamodel.vendors.VendorConversationUnique;
import com.xiaoyu.im.provider.ConversationManager;

import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import in.srain.cube.concurrent.AppThreads;
import in.srain.cube.concurrent.SimpleTask;
import in.srain.cube.util.CLog;
import in.srain.cube.util.TimeUtils;
import in.srain.cube.util.general.MapUtil;
import in.srain.cube.util.internal.AppFunction;
import in.srain.cube.util.internal.AppObserver;
import in.srain.cube.util.internal.AppSupplier;
import io.reactivex.Observable;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;

import static com.xiaoyu.im.client.proxy.AbstractProxy.QueryDirection.DirectionDef;

public abstract class AbstractProxy<VendorMessage, ConversationUnique extends VendorConversationUnique, Notification> implements IMProxy {

    private static final String TEMP_CHAT_ID = "temp_chat_id";

    protected abstract void vendorLogin(IMTokenInfo tokenInfo, IMCallback<Void> callback);

    protected abstract void vendorLogout();

    protected abstract MessageBridge<VendorMessage, ConversationUnique, Notification> getMessageBridge();

    protected abstract void sendVendorMessage(ConversationUnique unique, VendorMessage message, IMCallback<VendorMessage> callback);

    protected abstract void recallVendorMessage(ConversationUnique unique, VendorMessage message, IMCallback<Void> callback);

    protected abstract void observeReceiveVendorMessage(AppObserver<List<VendorMessage>> receiveMessageObserver);

    protected abstract void observeReceiveVendorNotification(AppObserver<Notification> observer);

    protected abstract void observeRecallVendorMessage(AppObserver<VendorMessage> recallMessageObserver);

    protected abstract void clearVendorUnreadCount(ConversationUnique conversationUnique);

    protected abstract int getVendorUnreadMessageCount(ConversationUnique unique);

    protected abstract void sendVendorMessageReceipt(VendorMessage message);

    protected abstract void queryVendorMessagesFromRemote(int limit, ConversationUnique unique, long fromTime, long toTime, @DirectionDef int direction, IMCallback<List<VendorMessage>> callback);

    private static final String TAG = "IMProxy";
    private static final ExpiringLogging LOGGING = new ExpiringLogging();
    public static AppFunction<AppMessage, AppMessage> sReceiveFunction;

    private final AtomicReference<IMTokenInfo> mSuccessIMTokenInfo = new AtomicReference<>(IMTokenInfo.EMPTY);
    private final String mVendorType;
    private AtomicBoolean mIsLoginSyncFinished = new AtomicBoolean(false);
    private Disposable mRetryReceiveMessageDisposable;

    protected AbstractProxy(@IMVendorType.IMVendorTypeDef String vendorType) {
        this.mVendorType = vendorType;
    }

    @IMVendorType.IMVendorTypeDef
    @Override
    public String getVendorType() {
        return mVendorType;
    }

    @Override
    @NonNull
    public IMTokenInfo getSuccessIMTokenInfo() {
        return mSuccessIMTokenInfo.get();
    }

    private <T, R> T callbackErrorIfNull(IMCallback<R> callback, AppSupplier<T> supplier, R result) {
        try {
            return Objects.requireNonNull(supplier.get());
        } catch (Throwable exception) {
            callback.onResult(false, result, new IMException(exception));
            CLog.w(IMConstant.LOG_TAG, "callbackErrorIfNull %s", Log.getStackTraceString(exception));
            return null;
        }
    }

    @Override
    public void sendMessage(AppMessage message, IMCallback<AppMessage> callback) {
        final VendorMessage vendorMessage = callbackErrorIfNull(callback, () -> getMessageBridge().create(message), null);
        if (vendorMessage == null) {
            return;
        }
        final ConversationUnique unique = callbackErrorIfNull(callback, () -> getMessageBridge().toConversationUnique(message.chatToken.getChatId()), null);
        if (unique == null) {
            return;
        }

        final IMCallback<AppMessage> toAppMessageCallback = (success, successMessage, exception) -> {
            if (successMessage != null) {
                successMessage.setDisplayTime(message.displayTime);
                callback.onResult(success, successMessage, exception);
            } else {
                callback.onResult(false, null, exception);
            }
        };

        final IMCallback<VendorMessage> vendorCallback = (success, successVendorMessage, exception) -> {
            if (success && successVendorMessage != null) {
                getMessageBridge().toAppMessage(successVendorMessage, toAppMessageCallback);
            } else {
                AppThreads.runOnIOThread(() -> callback.onResult(false, null, exception));
            }
        };
        sendVendorMessage(unique, vendorMessage, vendorCallback);
    }

    @Override
    public void sendMessageReceipt(AppMessage message) {
        final IMCallback<VendorMessage> recreateCallback = (success, vendorMessage, exception) -> {
            if (vendorMessage != null) {
                sendVendorMessageReceipt(vendorMessage);
            }
        };
        getMessageBridge().recreate(message, recreateCallback);
    }

    @Override
    public void recallMessage(AppMessage message, IMCallback<Void> callback) {
        final ConversationUnique unique = callbackErrorIfNull(callback, () -> getMessageBridge().toConversationUnique(message.chatToken.getChatId()), null);
        if (unique == null) {
            return;
        }

        final IMCallback<VendorMessage> recreateCallback = (success, vendorMessage, exception) -> {
            if (vendorMessage != null) {
                recallVendorMessage(unique, vendorMessage, callback);
            } else {
                callback.onResult(false, null, exception);
            }
        };
        getMessageBridge().recreate(message, recreateCallback);
    }

    @Override
    public void observeReceiveMessage(AppObserver<List<AppMessage>> receiveMessageObserver) {
        final AppObserver<List<VendorMessage>> vendorObserver = vendorMessages -> {
            AppThreads.runOnIOThread(() -> {
                ConversationWithMessageListWrapper wrapper = convertAndFilterReceiveMessagesToWrapper(vendorMessages);
                upsertReceiveMessages(UserData.getInstance().getUid(), wrapper.getMessages());

                // Only ensureConversation() immediately after delay 2s since Nim login sync completed,
                // before that, the single conversation's message will call observeReceiveMessage() 1 time
                if (mIsLoginSyncFinished.get()) {
                    List<AppMessage> messages = ConversationManager.getInstance().ensureConversationMapAndGetMessagesByVendor(wrapper.getConversationMessageListMap());
                    receiveMessageObserver.onEvent(messages);
                }
            });
        };
        observeReceiveVendorMessage(vendorObserver);
    }

    private void retryReceiveMessage() {
        if (!UserData.getInstance().hasLogin()) {
            return;
        }
        CLog.i(IMConstant.LOG_TAG, "retryReceiveMessage() start ensure");
        String uid = UserData.getInstance().getUid();
        Map<VendorConversationUnique, List<AppMessage>> map = new HashMap<>();
        List<AppMessage> messages = DBCall.call(uid, database -> ListUtil.convertToList(database.messageDao().getMessagesByChatId(TEMP_CHAT_ID, Integer.MAX_VALUE), messageEntity -> {
            AppMessage message = AppMessage.fromEntity(messageEntity);
            VendorConversationUnique unique = VendorConversationUnique.from(message.getVendorUniqueJson());
            bindConversationUniqueMessage(map, unique, message);
            return message;
        }));

        if (ListUtil.isNullOrEmpty(messages)) {
            return;
        }

        messages = ConversationManager.getInstance().ensureConversationMapAndGetMessagesByVendor(map);
        IMClient.getInstance().receiveMessages(messages);
    }

    private ConversationWithMessageListWrapper convertAndFilterReceiveMessagesToWrapper(List<VendorMessage> vendorMessages) {
        Map<VendorConversationUnique, List<AppMessage>> map = new HashMap<>();
        List<AppMessage> messages = ListUtil.convertToListFilterNull(vendorMessages, vendorMessage -> {
            try {
                AppMessage message = getMessageBridge().tryToAppMessage(vendorMessage, TEMP_CHAT_ID);
                message = filterMessage(message, ReceiveType.MESSAGE);
                if (message == null) {
                    return null;
                }

                VendorConversationUnique unique = getMessageBridge().toConversationUnique(vendorMessage);
                bindConversationUniqueMessage(map, unique, message);
                return message;
            } catch (Throwable t) {
                if (vendorMessage instanceof IMMessage) {
                    IMLog.logToAppMessageError("convertReceiveMessage", (IMMessage) vendorMessage, t);
                }
                t.printStackTrace();
                return null;
            }
        });
        return new ConversationWithMessageListWrapper(map, messages);
    }

    private void bindConversationUniqueMessage(Map<VendorConversationUnique, List<AppMessage>> map, VendorConversationUnique unique, AppMessage message) {
        if (unique == null) {
            return;
        }
        message.bindVendorUnique(unique);

        List<AppMessage> curMessageList = MapUtil.getOrDefault(map, unique, new CopyOnWriteArrayList<>());
        curMessageList.add(message);
    }

    private void upsertReceiveMessages(String uid, List<AppMessage> appMessages) {
        List<MessageEntity> entities = ListUtil.convertToList(appMessages, AppMessage::toEntity);
        List<Long> rowIds = DBCall.listCall(uid, database -> database.messageDao().upsertAll(entities));
        IMLog.logImStreamForUpsertMessages(appMessages, rowIds);
    }

    private void upsertReceiveMessage(String uid, AppMessage appMessage) {
        Long rowId = DBCall.call(uid, database -> database.messageDao().upsert(appMessage.toEntity()));
        IMLog.logImStreamForUpsertMessages(Collections.singletonList(appMessage), Collections.singletonList(rowId));
    }

    @Override
    public void observeCustomNotification(AppObserver<List<AppMessage>> customNotificationObserver) {
        AppObserver<Notification> observer = notification -> {
            AppThreads.runOnIOThread(() -> {
                Map<VendorConversationUnique, List<AppMessage>> map = new HashMap<>();
                try {
                    AppMessage message = getMessageBridge().tryNotificationToAppMessage(notification, TEMP_CHAT_ID);
                    message = filterMessage(message, ReceiveType.NOTIFICATION);
                    if (message == null) {
                        return;
                    }

                    VendorConversationUnique unique = getMessageBridge().notificationToConversationUnique(notification);
                    bindConversationUniqueMessage(map, unique, message);

                    upsertReceiveMessage(UserData.getInstance().getUid(), message);

                    if (mIsLoginSyncFinished.get()) {
                        List<AppMessage> messageList = ConversationManager.getInstance().ensureConversationMapAndGetMessagesByVendor(map);
                        customNotificationObserver.onEvent(messageList);
                    }
                } catch (Throwable throwable) {
                    if (notification instanceof CustomNotification) {
                        IMLog.logNotificationToAppMessageError("convert_custom_notification", (CustomNotification) notification, throwable);
                    }
                    throwable.printStackTrace();
                }
            });
        };
        observeReceiveVendorNotification(observer);
    }

    private AppMessage filterMessage(AppMessage message, String receiveType) {
        IMLog.logImStreamForFilterReceiveMessage(message, receiveType);
        if (message.invisible) {
            return null;
        }
        message = sReceiveFunction.invoke(message);
        return message;
    }

    @Override
    public void observeRecallMessage(AppObserver<AppMessage> recallMessageObserver) {
        final AppObserver<VendorMessage> vendorObserver = vendorMessage -> {
            AppThreads.runOnIOThread(() -> {
                final VendorConversationUnique unique = getMessageBridge().toConversationUnique(vendorMessage);
                ConversationManager.getInstance().ensureConversationByVendor(unique);
                final AppMessage appMessage = getMessageBridge().toAppMessageOrNull(vendorMessage);
                if (appMessage != null) {
                    recallMessageObserver.onEvent(appMessage);
                }
            });
        };
        observeRecallVendorMessage(vendorObserver);
    }

    @Override
    public int getUnreadMessageCount(String chatId) {
        final ConversationUnique unique = getMessageBridge().toConversationUnique(chatId);
        return unique != null ? getVendorUnreadMessageCount(unique) : 0;
    }

    @Override
    public void clearUnreadCount(String chatId) {
        final ConversationUnique unique = getMessageBridge().toConversationUnique(chatId);
        if (unique != null) {
            clearVendorUnreadCount(unique);
        }
    }

    @Override
    public void queryMessagesFromRemote(int limit, String chatId, long fromTime, long toTime, @DirectionDef int direction, IMCallback<List<AppMessage>> callback) {
        final ConversationUnique unique = getMessageBridge().toConversationUnique(chatId);
        final IMCallback<List<VendorMessage>> vendorCallback = createVendorCallbackForQueryMessages(callback);
        queryVendorMessagesFromRemote(limit, unique, fromTime, toTime, direction, vendorCallback);
    }

    @Nullable
    @Override
    public VendorConversationUnique toConversationUnique(String chatId) {
        return getMessageBridge().toConversationUnique(chatId);
    }

    private IMCallback<List<VendorMessage>> createVendorCallbackForQueryMessages(IMCallback<List<AppMessage>> callback) {
        return (success, vendorMessages, exception) -> {
            AppThreads.runOnIOThread(() -> {
                final List<AppMessage> messageList = ListUtil.convertToListFilterNull(vendorMessages, getMessageBridge()::toAppMessageOrNull);
                callback.onResult(success, messageList, exception);
            });
        };
    }

    public List<AppMessage> queryMessagesFromRemote(int limit, String chatId, long fromTime, long toTime, @DirectionDef int direction) throws Throwable {
        AtomicBoolean successReference = new AtomicBoolean();
        AtomicReference<List<AppMessage>> resultReference = new AtomicReference<>();
        AtomicReference<IMException> exceptionReference = new AtomicReference<>();

        CountDownLatch latch = new CountDownLatch(1);
        queryMessagesFromRemote(limit, chatId, fromTime, toTime, direction, (success, result, exception) -> {
            successReference.set(success);
            resultReference.set(result != null ? result : Collections.emptyList());
            exceptionReference.set(exception);
            latch.countDown();
        });
        latch.await();

        if (exceptionReference.get() != null) {
            throw exceptionReference.get();
        }
        if (!successReference.get()) {
            throw new RuntimeException("query messages from remote sync error");
        }
        return resultReference.get();
    }

    @Override
    public boolean isOnline() {
        final boolean online = IMUtil.isOnline(getOnlineStatus());
        CLog.d(IMConstant.LOG_TAG, "%s isOnline %s", getVendorType(), online);
        return online;
    }

    @Override
    public boolean shouldNotLoginAgain() {
        final boolean shouldNotLogin = IMUtil.shouldNotLogin(getOnlineStatus());
        CLog.d(IMConstant.LOG_TAG, "%s isOnline %s", getVendorType(), shouldNotLogin);
        return shouldNotLogin;
    }

    @Override
    public synchronized void login() {
        CLog.d(IMConstant.LOG_TAG, "%s login", getVendorType());
        if (shouldNotLoginAgain()) {
            CLog.d(IMConstant.LOG_TAG, "%s skip login because online", getVendorType());
            return;
        }
        if (!LOGGING.setLoggingIfNot()) {
            CLog.d(IMConstant.LOG_TAG, "%s skip login because logging", getVendorType());
            return;
        }
        ensureToken();
    }

    private synchronized void loginFromLocal() {
        CLog.d(IMConstant.LOG_TAG, "%s loginFromLocal", getVendorType());
        if (shouldNotLoginAgain()) {
            CLog.d(IMConstant.LOG_TAG, "%s skip loginFromLocal because online", getVendorType());
            LOGGING.setNotLogging();
            return;
        }

        final IMTokenInfo tokenInfo = IMTokenData.getTokenInfo(getVendorType());
        if (tokenInfo.isEmpty()) {
            CLog.e(IMConstant.LOG_TAG, "%s loginFromLocal error because token empty", getVendorType());
            ensureToken();
            return;
        }

        final IMCallback<Void> callback = (success, result, exception) -> {
            if (success) {
                CLog.i(IMConstant.LOG_TAG, "%s login success %s", getVendorType(), tokenInfo.vendorUid);
                mSuccessIMTokenInfo.set(tokenInfo);
                LOGGING.setNotLogging();
            } else {
                CLog.e(IMConstant.LOG_TAG, "%s login error %s %s", getVendorType(), tokenInfo.vendorUid, exception);

                SimpleTask.postDelay(this::ensureToken, 3 * TimeUtils.MILLIS_IN_SECOND);

                String desc = String.format("error=%s, detail=%s", exception.getMessage(), Log.getStackTraceString(exception));
                IMLog.logLoginError("manual_login", exception.code, desc);
                if (BuildConfig.DEBUG) {
                    GlobalUI.getInstance().showToast("lib_im: login " + desc);
                }
            }
        };
        vendorLogin(tokenInfo, callback);
    }

    private void ensureToken() {
        IMTokenData.ensureToken(getVendorType(), this::loginFromLocal);
    }

    @Override
    public synchronized void logout() {
        CLog.d(IMConstant.LOG_TAG, "%s logout", getVendorType());
        vendorLogout();
        LOGGING.setNotLogging();
        mSuccessIMTokenInfo.set(IMTokenInfo.EMPTY);
    }

    public void onFetchFinish(String chatId, List<AppMessage> messages) {
        final String uid = UserData.getInstance().getUid();
        final List<MessageEntity> entities = convertFetchMessageToEntity(messages);
        DBCall.run(uid, database -> database.messageDao().upsertAll(entities));
        CLog.d(IMConstant.LOG_TAG, "%s onFetchFinish size=%s", TAG, entities.size());
    }

    private List<MessageEntity> convertFetchMessageToEntity(final List<AppMessage> messages) {
        return ListUtil.convertToList(messages, appMessage -> {
            MessageEntity entity = appMessage.toEntity();
            entity.setSyncPoint(1);
            return entity;
        });
    }

    protected void startRetryReceiveMessage() {
        RxUtils.disposable(mRetryReceiveMessageDisposable);
        this.mRetryReceiveMessageDisposable = Observable.interval(2000, 6000, TimeUnit.MILLISECONDS)
                .observeOn(Schedulers.io())
                .subscribe(aLong -> {
                    retryReceiveMessage();
                    mIsLoginSyncFinished.set(true);
                });
    }

    private static class ExpiringLogging {

        private final AtomicBoolean mLogging = new AtomicBoolean(false);

        boolean setLoggingIfNot() {
            return mLogging.compareAndSet(false, true);
        }

        void setNotLogging() {
            mLogging.set(false);
        }
    }

    public static class QueryDirection {
        public static final int NEW = 1;
        public static final int OLD = 2;

        @IntDef({NEW, OLD})
        public @interface DirectionDef {
        }
    }
}
